package com.epoint.flinkdemo.process;

import com.epoint.flinkdemo.entity.SenSorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import scala.Tuple3;

public class AlarmProcessFunction extends ProcessFunction<SenSorReading, Tuple3<String, Double, Double>> {

    private transient ValueState<Double> valueState;

    private Double threshold;

    public AlarmProcessFunction(Double threshold) {
        this.threshold = threshold;
    }

    @Override
    public void processElement(SenSorReading value, Context ctx, Collector<Tuple3<String, Double, Double>> out) throws Exception {
        Double beforeValue = valueState.value();
        if (beforeValue != null) {
            Double diff = Math.abs(value.getTemp() - beforeValue);
            if (diff > threshold) {
                out.collect(new Tuple3<>(value.getId(), beforeValue, value.getTemp()));
            }
        }
        // 更新最新的值
        valueState.update(value.getTemp());
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("alarm-temp", Double.class));
    }

    @Override
    public void close() throws Exception {
        valueState.clear();
    }
}
